package com.gitee.sparta.core.listener;

import com.gitee.sparta.core.bean.cluster.ClusterCmd;
import com.gitee.sparta.common.IdentityEnum;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.bean.cluster.ServerBean;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.process.feature.SpartaWorkerFeature;
import com.gitee.sparta.core.process.operate.AbstractTimeoutTabOperate;
import com.gitee.sparta.core.process.workers.SpartaCarryWorker;
import com.gitee.sparta.core.process.workers.SpartaListenWorker;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.channel.Channel;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;

import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.Objects;

/**
 * <p>
 * sparta 处理宕机的策略
 *   出现的场景分析：
 *      此时服务器正在运行，readable_tab 中包含已读取({@see SpartaConstant#HAVE_READ})和未读取的数据
 *      下一刻，关闭了服务，此时的 readable_tab 中还是有这些数据，然后消费者与消费者之间是隔离的，也就是
 *      说，如果订单服务 A 机器宕机了，订单服务 B 机器是不会消费 A 机器的数据的，所以这里需要将数据重新放
 *      入到 timeout_tab 中，提供 B 服务继续消费
 *
 * spring 关机流程: {@see AbstractApplicationContext#doClose()}
 *  1. 发布一个 ContextCloseEvent 事件   {@see publishEvent(new ContextClosedEvent(this))}
 *  2. 关闭所有的 bean, 实际上就是从 IOC 容器中删除 bean  {@see AbstractApplicationContext#destroyBeans()}
 *  3. 关闭创建 bean 的工厂, 实际上是为 serializationId 设置 null 值 {@see AbstractApplicationContext#closeBeanFactory()}
 * </p>
 *
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
@AllArgsConstructor
public class SpartaClosedListener<V> implements ApplicationListener<ContextClosedEvent> {

    private final RedissonClient client;
    private final SpartaCoreProperties properties;
    private final AbstractTimeoutTabOperate<V> timeoutTabOperate;
    private final SpartaCarryWorker<V> spartaCarryWorker;
    private final SpartaListenWorker<V> spartaListenWorker;

    /**
     * 注意，必须按照如下的顺序进行关闭
     *
     * @param event     spring event
     */
    @SuppressWarnings("all")
    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        // 禁止参与候选
        SpartaWorkerFeature.FORBID_CANDIDATE = true;

        // 1. 关闭与 leader 的连接
        if (Objects.equals(ClusterCache.getServerIdentity(), IdentityEnum.FOLLOWER)) {
            // 这条消息不一定能发出去
            ClusterProtocol protocol = new ClusterProtocol();
            protocol.setCmd(ClusterCmd.FOLLOWER_SHUTDOWN);
            protocol.setServerId(ClusterCache.getServerId());
            protocol.setLeaderSid(ClusterCache.getLeaderSid());
            AbstractClusterFollower.sendQueue(protocol);

            // 所以采取更为妥当的方案
            if (ClusterCache.isNetty()) {
                Channel channel = ClusterCache.getChannel(this.properties.getIsolation());
                if (Objects.nonNull(channel)) {
                    channel.close();
                    log.info("[Sparta] Netty close follower channel now ...... ");
                }
            } else {
                SocketChannel channel = ClusterCache.getClient(this.properties.getIsolation());
                try {
                    if (Objects.nonNull(channel)) {
                        channel.close();
                        log.info("[Sparta] Nio close follower channel now ...... ");
                    }
                } catch (IOException e) {
                    log.error("[Sparta] Close follower channel error.", e);
                }
            }
        }

        // 2. 服务注销
        this.serviceShutdown();
        log.info("[Sprata] Server shutdown now ...... ");

        // 3. 关闭异步处理的线程，窃取工作
        this.timeoutTabOperate.shutdown();
        log.info("[Sparta] Closed async write thread ..");

        // 4. 关闭搬运线程
        this.spartaCarryWorker.shutdown();
        log.info("[Sparta] Closed carry thread ..");

        // 5. 关闭监听线程池
        this.spartaListenWorker.shutdown();
        log.info("[Sparta] Closed listen threads ..");

        // 6. 将 readable_tab 中全部数据放入到 timeout_tab 中，回滚数据
        if (this.properties.isEnableCloseRollback()) {
            this.timeoutTabOperate.closeRollback();
            log.info("[Sparta] Rolled back no consumption topic");
        }
    }

    /**
     * 服务关闭后，需要把服务从注册表中删除
     */
    @SuppressWarnings("all")
    private void serviceShutdown () {
        RMap<String, ServerBean> serverTable = this.client.getMap(this.properties.getServerTable());
        serverTable.remove(ClusterCache.getServerId());
    }
}
